package lyw.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import lyw.app.BaseAppV2;
import lyw.bean.funnelActionBean;
import lyw.util.AtguiguUtil;
import lyw.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.HashMap;

import static lyw.common.Constant.*;

/**
 * @Author: lyw
 * @Date: 2022/9/1 21:25
 * @Describe:
 */
public class Dws_03_DwsUserUserActionWindow extends BaseAppV2 {
    public static void main(String[] args) {
        new Dws_03_DwsUserUserActionWindow().init(
                4003,
                2,
                "Dws_03_DwsUserUserActionWindow",
                TOPIC_DWD_TRAFFIC_PAGE,
                TOPIC_DWD_TRADE_CART_ADD,
                TOPIC_DWD_TRADE_USER_ORDER_DETAIL
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          HashMap<String, DataStreamSource<String>> streams) {
        streams.get(TOPIC_DWD_TRAFFIC_PAGE).print();
        // 1. 把里转成同一种类型, 然后union成一个流
        System.out.println("---------1-------");
        DataStream<funnelActionBean> beanStream = parseAndUnionOne(streams);
        // 2. 开窗聚合
        System.out.println("-------------2---------------");
        SingleOutputStreamOperator<funnelActionBean> resultStream = windowAndAgg(beanStream);
//                resultStream.print("normal");
        //        resultStream.getSideOutput(new OutputTag<TrafficPageViewBean>("late") {}).print("late");
        // 3. 写出到doris中
        writeToDoris(resultStream);
    }


    private void writeToDoris(SingleOutputStreamOperator<funnelActionBean> resultStream) {
        resultStream
                .map(bean -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
//                config.propertyNamingStrategy = PropertyNamingStrategy.KebabCase;  // 转成json的时候, 属性名使用下划线
                    String json = JSON.toJSONString(bean, config);
                    System.out.println(json);
                    return json;
                })

                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_user_user_action_window"));
    }

    private SingleOutputStreamOperator<funnelActionBean> windowAndAgg(DataStream<funnelActionBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<funnelActionBean>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(funnelActionBean::getUid)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<funnelActionBean>("late") {})
                .reduce(
                        new ReduceFunction<funnelActionBean>() {
                            @Override
                            public funnelActionBean reduce(funnelActionBean bean1,
                                                           funnelActionBean bean2) throws Exception {
                                bean1.setPvCtHome(bean1.getPvCtHome() + bean2.getPvCtHome());
                                bean1.setPvCtDetail(bean1.getPvCtDetail() + bean2.getPvCtDetail());
                                bean1.setAddCart(bean1.getAddCart() + bean2.getAddCart());
                                bean1.setOrderNum(bean1.getOrderNum() + bean2.getOrderNum());
                                bean1.setPaySuc(bean1.getPaySuc() + bean2.getPaySuc());
                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<funnelActionBean, funnelActionBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<funnelActionBean> elements, // 有且仅有一个值: 聚合函数中最终一的结果
                                                Collector<funnelActionBean> out) throws Exception {
                                funnelActionBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));

                                bean.setCurDate(AtguiguUtil.toDate(System.currentTimeMillis())); // 统计日期

                                out.collect(bean);
                            }
                        }
                );
    }

    private DataStream<funnelActionBean> parseAndUnionOne(HashMap<String, DataStreamSource<String>> streams) {

        //1.page
        SingleOutputStreamOperator<funnelActionBean> pvHomeDetailStream = streams.get(TOPIC_DWD_TRAFFIC_PAGE)
                .map(JSON::parseObject)
                .filter(obj -> ("home".equals(obj.getJSONObject("page").getString("page_id")) || "course_detail".equals(obj.getJSONObject("page").getString("page_id"))))  //todo 只要首页或详情页数据
                .keyBy(json ->json.getString("uid"))    //todo 使用键控状态，必须要使用keyby（）分组
                .process(new KeyedProcessFunction<String,JSONObject, funnelActionBean>() {

                    private ValueState<String> visitDetailDateState;

                    private ValueState<String> visitHomeDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        visitHomeDateState =
                                getRuntimeContext().getState(new ValueStateDescriptor<String>("visitHomeDateState", String.class));
                        visitDetailDateState =
                                getRuntimeContext().getState(new ValueStateDescriptor<String>("visitDetailDateState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject obj,
                                               Context ctx,
                                               Collector<funnelActionBean> out) throws Exception {

                        String homeDate = visitHomeDateState.value();
                        String detailDate = visitDetailDateState.value();

                        String today = AtguiguUtil.toDate(System.currentTimeMillis());

                        JSONObject common = obj.getJSONObject("common");
                        JSONObject page = obj.getJSONObject("page");

                        if (!today.equals(homeDate) || !today.equals(detailDate)) {  //todo 今天第一次登陆

                            String uid = common.getString("uid");

                            // todo 首页浏览人数
                            Long pvCtHome = 0L;
                            if (page.getString("page_id") == "home"){
                                pvCtHome = 1L;
                                visitHomeDateState.update(today);
                            }

                            // todo 商品详情页浏览人数
                            Long pvCtDetail = 0L;
                            if (page.getString("page_id") == "course_detail"){
                                pvCtDetail = 0L;
                                visitDetailDateState.update(today);
                            }

                            //加购人数
                            Long addCart = 0L;
                            //下单人数
                            Long orderNum = 0L;
                            //支付成功人数
                            Long paySuc = 0L;

                            Long ts = obj.getLong("ts");
                            out.collect(new funnelActionBean(
                                    "", "",
                                    uid,
                                    "",               //统计日期
                                    pvCtHome, pvCtDetail, addCart, orderNum, paySuc,  //指标列
                                    ts
                            ));
                            visitHomeDateState.update(today);// 更新状态
                        }
                    }
                });

        //2.addCart
        SingleOutputStreamOperator<funnelActionBean> addCartStream = streams.get(TOPIC_DWD_TRADE_CART_ADD)
                .map(JSON::parseObject)
                .keyBy(json ->json.getString("user_id"))
                .process(new KeyedProcessFunction<String,JSONObject, funnelActionBean>() {

                    private ValueState<String> addCartDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
//                        System.out.println("addCartStream" + "==========");
                        addCartDateState =
                                getRuntimeContext().getState(new ValueStateDescriptor<String>("addCartDateState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject obj,
                                               Context ctx,
                                               Collector<funnelActionBean> out) throws Exception {
//                        System.out.println("addCartStream---------processElement");
                        String addCartDate = addCartDateState.value();
                        String today = AtguiguUtil.toDate(System.currentTimeMillis());

                        if (!today.equals(addCartDate)) {  //todo 今天第一次登陆

                            String uid = obj.getString("user_id");

                            // 首页浏览人数
                            Long pvCtHome = 0L;
                            // 商品详情页浏览人数
                            Long pvCtDetail = 0L;

                            //加购人数
                            Long addCart = 1L;
                            //下单人数
                            Long orderNum = 0L;
                            //支付成功人数
                            Long paySuc = 0L;

                            Long ts = obj.getLong("ts");
                            out.collect(new funnelActionBean(
                                    "", "",
                                    uid,
                                    "",               //统计日期
                                    pvCtHome, pvCtDetail, addCart, orderNum, paySuc,  //指标列
                                    ts
                            ));
                            addCartDateState.update(today);// 更新状态
                        }
                    }
                });

        //3.order payment
        SingleOutputStreamOperator<funnelActionBean> orderPaymentStream = streams.get(TOPIC_DWD_TRADE_USER_ORDER_DETAIL)
                .map(JSON::parseObject)
                .filter(obj -> ("1002".equals(obj.getString("order_status")) || "1003".equals(obj.getString("order_status")) ))  //todo 1002 下单   1003 已支付
                .keyBy(json ->json.getString("user_id"))
                .process(new KeyedProcessFunction<String,JSONObject, funnelActionBean>() {


                    private ValueState<String> paymentDateState;
                    private ValueState<String> orderDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        orderDateState =
                                getRuntimeContext().getState(new ValueStateDescriptor<String>("orderDateState", String.class));
                        paymentDateState =
                                getRuntimeContext().getState(new ValueStateDescriptor<String>("paymentDateState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject obj,
                                               Context ctx,
                                               Collector<funnelActionBean> out) throws Exception {

                        String orderDate = orderDateState.value();
                        String paymentDate = paymentDateState.value();

                        String today = AtguiguUtil.toDate(System.currentTimeMillis());

                        if (!today.equals(orderDate) || !today.equals(paymentDate)) {  //todo 今天第一次登陆

                            String uid = obj.getString("user_id");

                            // 首页浏览人数
                            Long pvCtHome = 0L;
                            // 商品详情页浏览人数
                            Long pvCtDetail = 0L;

                            //加购人数
                            Long addCart = 0L;
                            //下单人数
                            Long orderNum = 0L;
                            if ("1002".equals(obj.getString("order_status"))){
                                orderNum = 1L;
                                orderDateState.update(today);
                            }
                            //支付成功人数
                            Long paySuc = 0L;
                            if ("1003".equals(obj.getString("order_status"))){
                                paySuc = 1L;
                                paymentDateState.update(today);
                            }
                            Long ts = obj.getLong("ts");
                            out.collect(new funnelActionBean(
                                    "", "",
                                    uid,
                                    "",               //统计日期
                                    pvCtHome, pvCtDetail, addCart, orderNum, paySuc,  //指标列
                                    ts
                            ));
                        }
                    }
                });

        return pvHomeDetailStream.union(addCartStream,orderPaymentStream);
    }

}
